/**
 * Copyright (c) 2020 kedacom
 * OpenATC is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 * http://license.coscl.org.cn/MulanPSL2
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
 **/

package com.openatc.agent.handler;

import com.google.gson.Gson;
import com.google.gson.JsonObject;
import com.openatc.agent.controller.TemplateController;
import com.openatc.agent.gateway.WebSocketServer;
import com.openatc.agent.model.Fault;
import com.openatc.agent.service.AscsDao;
import com.openatc.agent.service.DevService;
import com.openatc.agent.service.FaultService;
import com.openatc.agent.service.HistoryDataDao;
import com.openatc.comm.data.MessageData;
import com.openatc.comm.handler.ICommHandler;
import com.openatc.core.model.RESTRet;
import com.openatc.core.util.DateUtil;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import javax.websocket.Session;
import java.io.ObjectInputStream;
import java.util.Date;
import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.logging.Logger;

import static com.openatc.comm.common.CosntDataDefine.detectorstatus;
import static com.openatc.comm.common.CosntDataDefine.login;

/**
 * @ClassName AgentHandler
 * @Auther zhangwenchao
 * @CreateTime 2019/11/5 16:09
 * @Description
 */
@Component
public class AgentHandler extends ICommHandler {

    private static Logger logger = Logger.getLogger(AgentHandler.class.toString());

    @Value("${spring.redis.enable}")
    private boolean isRedisEnable;

    @Autowired
    public StringRedisTemplate stringRedisTemplate;
    @Autowired
    private FaultService faultService;
    @Autowired
    private AscsDao ascsDao;
    @Autowired
    private HistoryDataDao historyDataDao;
    @Autowired
    private DevService devService;
    @Autowired
    private TemplateController templateController;

    public static BlockingQueue<MessageData> MdQueue = new LinkedBlockingQueue<>();

    public static int MdTodayCount = 0; // 今日收到消息数
    public static int MdminQps = 0; // 5分钟QPS
    public static Queue<String> patternsqlqueue = new ConcurrentLinkedQueue<>(); // 批量保存的方案sql缓存

    private String agenttype = "asc";

    private Gson gson = new Gson();
    Map<Session, WebSocketServer> patternWebSocketSet = WebSocketServer.getPatternWebSocketSet();

    @Override
    public void process(MessageData msg) {

        try {
            MdQueue.put(msg);
            MdTodayCount++;
            MdminQps++;
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    @Async("taskExecutor")
    public void processExec() {
        while (true) {
            try {
                MessageData msg = null;
                try {
                    msg = MdQueue.take();
                } catch (InterruptedException e) {
                    continue;
                }
                // 如果消息为空，则返回
                if (msg == null) {
                    logger.warning("AgentHandler process: MessageData is null");
                    continue;
                }

                // 检查消息时间
                if (msg.getCreatetime() == null) {
                    msg.setCreatetime(DateUtil.date2esstr(new Date()));
                }

                // 设置系统收到时间
                msg.setSystime(DateUtil.date2esstr(new Date()));

                // 获取消息类型
                String infotype = msg.getInfotype();
                if (infotype == null) {
                    logger.warning("AgentHandler process: infoType is null");
                    continue;
                }

                // 获取操作类型
                String operation = msg.getOperation();

                String agentid;

                // 设备的注册消息
                if (login.equals(infotype)) {
                    agentid = ascsDao.processLoginMsg(msg);
                    if (agentid == null) continue;
                    // 把消息放入定时器的消息队列
//                    ascsDao.loginMap.put(agentid,msg);
                    AscsDao.putMsgToLoginMap(agentid,msg);
                    // redis发布
                    String loginKey = agenttype + ":" + infotype + ":" + agentid;
                    stringRedisTemplate.convertAndSend(loginKey,gson.toJson(msg));
                } else {
                    if (msg.getAgentid() != null) {
                        agentid = msg.getAgentid();
                    } else {
                        agentid = ascsDao.getAgentidFromThirdPartyid(msg.getThirdpartyid());
                        msg.setAgentid(agentid);
                    }
                }


                // ** 以下为消息的特殊处理 **
                // 程序运行到此处，agentid仍为null，说明在注册设备完成之前收到了其他上报的消息，丢弃
                if (agentid == null) continue;
                String key = agenttype + ":" + infotype + ":" + agentid;
                // 收到方案消息，发布redis订阅，并根据用户权限，发布ws消息
                if (infotype.equals("status/pattern")) {
                    saveAndPublishMsg(msg,key);
                    String value = gson.toJson(msg);

                    for (Session session : patternWebSocketSet.keySet()) {
                        WebSocketServer webSocketServer = patternWebSocketSet.get(session);
                        Set<String> agentIds = webSocketServer.patternAgentidSet;
                        if (agentIds != null && agentIds.contains(agentid)) {
                            webSocketServer.sendMessage(value);
                        }
                    }
                }
                // 上报的检测器状态
                else if ( infotype.equals(detectorstatus) ){
                    try {
                        msg = devService.processDetectorStatus(msg, key);
                    } catch (Exception e) {
                        logger.warning("preProcess " + msg.getAgentid() + "'s detector status error! " + e.getMessage());
                        continue;
                    }
                    String value = gson.toJson(msg);
                    stringRedisTemplate.opsForValue().set(key, value);
                }
                else if(infotype.equals("status/channel")){ // 通道状态消息，不处理
                    continue;
                }

                // 收到故障消息，发布故障消息，并保存到数据库中
                else if (infotype.equals("status/fault")) {
                    // 写入数据前，判断故障是否已处理或者已忽略，若是则不写入Redis,直接从MSg对象中删除这部分数据
                    msg = faultService.filterMsg(msg);
                    Fault[] m_faultDeque = gson.fromJson(msg.getData().getAsJsonObject().getAsJsonArray("m_FaultDeque"), Fault[].class);
                    if (m_faultDeque.length == 0)
                        continue;
                    else
                        faultService.processFaultMessage(msg);
                    saveAndPublishMsg(msg,key);
                }
                // 收到流量消息，保存到数据库中
                else if (infotype.equals("status/currentvolume")) {
                    historyDataDao.SaveFlowData(msg);
                    // 计算路口和相位统计数据，保存到redis以及数据库
                    processFlowData(msg);
                }
            } catch (Exception e) {
                logger.warning("taskExecutor Exception: " +  e.getMessage());
            }
        }

    }


    /**
     * 将消息保存到redis，并发布
     * @param msg
     * @param key
     */
    private void saveAndPublishMsg(MessageData msg, String key) {
        if (msg == null) return;
        // 消除对传入msg的影响
        MessageData newMsg = gson.fromJson(gson.toJson(msg),MessageData.class);
        if (isRedisEnable) {
            String value = gson.toJson(newMsg);
            stringRedisTemplate.opsForValue().set(key, value);
            stringRedisTemplate.convertAndSend(key, value);
        }

        String savePatternSql = getSavePatternSql(newMsg);
        if (StringUtils.hasLength(savePatternSql)) {
            patternsqlqueue.offer(savePatternSql);

        }
    }


    /**
     *  @param msg
     *
     */
    private void processFlowData(MessageData msg) {
        String agentid = msg.getAgentid();
        // 获取相位参数、检测器参数
        RESTRet<JsonObject> restRet;
        try {
            restRet = (RESTRet<JsonObject>)templateController.getTemplate(agentid);
        } catch (Exception e) {
            logger.warning(agentid + " request feature Exception: " + e.getMessage() + "，Update statistics data failed!");
            return;
        }

        if (restRet.isSuccess()) {
            try {
                JsonObject feature = restRet.getData();
                devService.savePhasesAndInteStatistics(feature,msg);
            } catch (Exception e){
                e.printStackTrace();
                logger.warning(agentid + " save statistics data Exception: " + e.getMessage());
            }
        } else {
            logger.warning("Request feature error: " + restRet.toString());
        }
    }

    /**
     * 拼接保存方案的sql
     * @param msg
     * @return
     */
    private String getSavePatternSql(MessageData msg) {
        JsonObject msgdata = null;

        try {
            // 删除不需要保存的数据
            msgdata = msg.getData().getAsJsonObject();

//            msgdata.remove("phase");
            msgdata.remove("overlap");
            msgdata.remove("rings");
            msgdata.remove("stages");

        } catch (Exception e){
            logger.warning("Deal Pattern Data error!:" + e.getMessage());
            return null;
        }

        String sql = String.format("INSERT INTO pattern_realtime (agentid,time,data) VALUES " +
                        "('%s','%s','%s')",
                msg.getAgentid(),msg.getCreatetime(),msgdata.toString());
        return sql;
    }

}
